package cn.ac.iscas.base.utils;

import cn.ac.iscas.domain.dameng.DamengColumnType;
import cn.ac.iscas.domain.flink.FlinkColumnType;
import cn.ac.iscas.domain.source.FlinkSourceColumn;

import java.util.List;

/**
 * @author LJian
 * @version 1.0
 * @description: TODO
 * @date 2022/6/17 17:54
 */
public class TypeConvertUtil {
    public static FlinkColumnType mysqlTypeToFlinkType(String mysqlType) {
        mysqlType = mysqlType.toLowerCase();
        switch (mysqlType) {
            case "tinyint":
                return FlinkColumnType.TINYINT;
            case "smallint":
            case "tinyint unsigned":
                return FlinkColumnType.SMALLINT;
            case "mediumint":
            case "int":
            case "integer":
            case "smallint unsigned":
                // 暂时这些类型统一认为是integer
                return FlinkColumnType.INT;
            case "bigint":
            case "int unsigned":
                return FlinkColumnType.BIGINT;
            case "bigint unsigned":
                return FlinkColumnType.DECIMAL_INT;
            case "float":
                return FlinkColumnType.FLOAT;
            case "double":
            case "double precision":
                return FlinkColumnType.DOUBLE;
            case "boolean":
                return FlinkColumnType.BOOLEAN;
            case "date":
                return FlinkColumnType.DATE;
            case "time":
                return FlinkColumnType.TIME;
            case "year":
            case "datetime":
            case "timestamp":
                // 暂时认为这些都是Date类型
                return FlinkColumnType.TIMESTAMP;
            case "char":
            case "varchar":
            case "tinytext":
            case "text":
            case "mediumtext":
            case "longtext":
                // 暂时认为这些都是字符串类型
                return FlinkColumnType.STRING;
            case "binary":
            case "tinyblob":
            case "blob":
            case "mediumblob":
            case "longblob":
                // 暂时认为这些都是字节类型
                return FlinkColumnType.BYTES;

            default:
                return FlinkColumnType.UNKOWN;
                //throw new DbRuntimeException(DbErrorCodes.UNSUPPORTED_DATA_TYPE_ERROR, String.format("不支持的MySQL数据类型:[%s]", mysqlType));
        }
        //return FlinkColumnType.UNKOWN;
    }

    public static String mysqlTypeToFlinkTypeString(String mysqlType) {
        return mysqlTypeToFlinkType(mysqlType).getName();
    }


    public static DamengColumnType toDamengType(String type) {
        switch (type.toLowerCase()) {
            case "tinyint":
                return DamengColumnType.TINYINT;
            case "smallint":
            case "tinyint unsigned":
                return DamengColumnType.SMALLINT;
            case "mediumint":
            case "int":
            case "integer":
            case "smallint unsigned":
            case "int unsigned":
                // 暂时这些类型统一认为是integer
                return DamengColumnType.INT;
            case "bigint":
            case "bigint unsigned":
                return DamengColumnType.BIGINT;
            case "float":
                return DamengColumnType.FLOAT;
            case "double":
            case "double precision":
                return DamengColumnType.DOUBLE;
            case "boolean":
                return DamengColumnType.BOOLEAN;
            case "date":
                return DamengColumnType.DATE;
            case "time":
                return DamengColumnType.TIME;
            case "year":
            case "datetime":
            case "timestamp":
                // 暂时认为这些都是Date类型
                return DamengColumnType.TIMESTAMP;
            case "char":
                return DamengColumnType.CHAR;
            case "varchar":
            case "string":
                return DamengColumnType.VARCHAR;
            case "tinytext":
            case "text":
            case "mediumtext":
            case "longtext":
                // 暂时认为这些都是字符串类型
                return DamengColumnType.TEXT;
            case "binary":
            case "tinyblob":
            case "blob":
            case "mediumblob":
            case "longblob":
                // 暂时认为这些都是字节类型
                return DamengColumnType.BLOB;

            default:
                return null;
            //throw new DbRuntimeException(DbErrorCodes.UNSUPPORTED_DATA_TYPE_ERROR, String.format("不支持的MySQL数据类型:[%s]", mysqlType));
        }
        //return FlinkColumnType.UNKOWN;
    }

    public static String columnsToSchemaString(List<FlinkSourceColumn> columns) {
        if (columns == null || columns.isEmpty()) {
            return null;
        }
        StringBuffer sb = new StringBuffer();
        for (FlinkSourceColumn column : columns){
            sb.append(String.format(" %s %s, ", column.getName(), column.getType()));
        }
        String tmp = sb.toString();
        tmp = tmp.substring(0, tmp.lastIndexOf(","));
        return tmp;
    }

    public static String columnsNameToSelectString(List<FlinkSourceColumn> columns) {
        if (columns == null || columns.isEmpty()) {
            return null;
        }
        StringBuffer sb = new StringBuffer();
        for (FlinkSourceColumn column : columns){
            sb.append(String.format(" %s , ", column.getName()));
        }
        String tmp = sb.toString();
        tmp = tmp.substring(0, tmp.lastIndexOf(","));
        return tmp;
    }

    public static String columnsToSelectString(List<FlinkSourceColumn> columns) {
        if (columns == null || columns.isEmpty()) {
            return null;
        }
        StringBuffer sb = new StringBuffer();
        for (FlinkSourceColumn column : columns){
            sb.append(String.format(" %s , ", column.getName()));
        }
        String tmp = sb.toString();
        tmp = tmp.substring(0, tmp.lastIndexOf(","));
        return tmp;
    }

}
